#include <pthread.h>
#include <malloc.h>
#include <unistd.h>

#include "msg.h"
#include "log.h"
#include "task.h"

struct msg_queue *msg_queue_create(void)
{
	log_info("begin msg_queue_create.");
	struct msg_queue *msg_queue = (struct msg_queue *)malloc(sizeof(struct msg_queue));	
	if (msg_queue == NULL) {
		log_error("No enough mem.");
		return NULL;
	}

	msg_queue->head = NULL;
	msg_queue->count = 0;

	int err = 0;
	err = pthread_mutex_init(&(msg_queue->lock), NULL);

	if (err != 0) {
		log_error("Cann't init mutex lock.");
		free(msg_queue);
		return NULL;
	}

	return msg_queue;
}

static int msg_queue_add_msg(struct msg_queue *msg_queue, struct msg *msg)
{
	log_info("begin msg_queue_add_msg.");
	if (msg_queue == NULL || msg == NULL) {
		log_warning("msg_queue is NULL or msg is NULL.");
		return 0;
	}

	pthread_mutex_lock(&(msg_queue->lock));
	struct msg *tmp;
	tmp = msg_queue->head;
	if (tmp == NULL) {
		msg->next = NULL;
		msg_queue->head = msg;
		msg_queue->count++;
	} else {
		for ( ; tmp->next != NULL; tmp = tmp->next) {
			/* waitting... */
		}

		msg->next = NULL;
		tmp->next = msg;
		msg_queue->count++;
	}
	pthread_mutex_unlock(&(msg_queue->lock));
	return 0;
}

static int msg_queue_del_msg(struct msg_queue *msg_queue, struct msg **msg)
{
	log_info("begin msg_queue_del_msg.");
	if (msg_queue == NULL || msg == NULL) {
		log_warning("msg_queue is NULL or msg is NULL.");
		return 0;
	}

	pthread_mutex_lock(&(msg_queue->lock));
	struct msg *tmp;
	tmp = msg_queue->head;

	while (tmp == NULL) {
		int ut = core_conf.usleep_time <= 0 ? 10 : core_conf.usleep_time;	
		pthread_mutex_unlock(&(msg_queue->lock));
		usleep(ut);
		pthread_mutex_lock(&(msg_queue->lock));
		tmp = msg_queue->head;
	}
	msg_queue->head = tmp->next;
	msg_queue->count--;
	*msg = tmp;
	pthread_mutex_unlock(&(msg_queue->lock));
	return 0;
}

int send_msg(struct msg *msg)
{
	log_info("begin send_msg.");
	if (msg == NULL) {
		log_error("msg is NULL.");
		return -1;
	}

	struct task *task_target = NULL;
	task_target = get_task_by_name(msg->to);
	if (task_target == NULL) {
		log_warning("cann't find task.");
		return -2;
	}

	int ret = 0;
	if (msg->is_request == 1) {
		ret = msg_queue_add_msg(task_target->msg_queue_request, msg);
	} else {
		ret = msg_queue_add_msg(task_target->msg_queue_result, msg);
	}

	return ret;
}

int receive_msg(struct msg **msg)
{
	log_info("begin receive_msg.");
	if (msg == NULL) {
		log_error("msg is NULL.");
		return -1;
	}

	struct task *task_target = NULL;
	pthread_t tid = pthread_self();
	task_target = get_task_by_tid(tid);
	if (task_target == NULL) {
		log_warning("cann't find task.");
		return -2;
	}

	if (task_target->msg_queue_result->head == NULL) {
		log_warning("no msg.");
		return -3;
	}

	int ret = 0;
	ret = msg_queue_del_msg(task_target->msg_queue_result, msg);

	return ret;
}

int get_request_msg(struct msg **msg, char *task_name)
{
	log_info("begin get_request_msg.");
	if (msg == NULL || task_name == NULL) {
		log_error("msg is NULL or task_name is NULL.");
		return -1;
	}

	struct task *task_target = NULL;
	task_target = get_task_by_name(task_name);
	if (task_target == NULL) {
		log_warning("cann't find task.");
		return -2;
	}

	if (task_target->msg_queue_request->count == 0) {
		log_info("no msg.");
		return -3;
	}

	int ret = 0;
	ret = msg_queue_del_msg(task_target->msg_queue_request, msg);

	return ret;
}
